package com.ai.wxy.spring.storm.test;

import com.ai.wxy.spring.storm.test.bolt.SplitBolt;
import com.ai.wxy.spring.storm.test.bolt.SumBolt;
import com.ai.wxy.spring.storm.test.spout.FileSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import java.util.concurrent.TimeUnit;

/**
 * 启动类
 *
 * @author 石头
 * @Date 2019/12/4
 * @Version 1.0
 **/
public class WordCountTopology {
    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("filespout",new FileSpout(),2);
        builder.setBolt("splitbolt",new SplitBolt(),5).shuffleGrouping("filespout");
        builder.setBolt("sumbolt",new SumBolt(),5).fieldsGrouping("splitbolt",new Fields("word"));

        Config config = new Config();
        config.setDebug(true);
        config.setNumWorkers(5);
        if (args!=null && args.length > 0){
            StormSubmitter.submitTopology(args[0],config,builder.createTopology());
        }else {
            // 本地模式
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("WordCountTopology", config, builder.createTopology());
        }

        TimeUnit.MINUTES.sleep(5);
    }
}
